package cn.edu.njpi.utils;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;

import java.io.IOException;
import java.net.URI;

/**
 * @author 朱宗辉
 * @version V1.0
 * @Package cn.edu.njpi.utils
 * @date 2021/11/17 下午8:11
 */
public class HDFSUtils {
    static String hdfsUri = "hdfs://192.168.35.31:9000";
    static Configuration conf = new Configuration();

    static {
        conf.set("fs.defaultFS", "hdfs://hadoop101:9000");
        System.setProperty("HADOOP_USER_NAME", "root");
    }

    // 上传文件
    public static void uploadInputFile(String localFile, String hdfsFile) throws IOException, InterruptedException {

        // FileSystem fs = FileSystem.get(configuration);

        String hdfsInput = hdfsUri + hdfsFile;
        FileSystem fs = FileSystem.get(URI.create(hdfsUri), conf, "root");
        fs.copyFromLocalFile(new Path(localFile), new Path(hdfsInput));
        fs.close();
    }

    // 下载文件
    public static void getOutput(String hdfsFile, String outputfile) throws IOException, InterruptedException {
        String remoteFile = hdfsUri + hdfsFile;
        Path path = new Path(remoteFile);
        FileSystem fs = FileSystem.get(URI.create(hdfsUri), conf, "root");
        fs.copyToLocalFile(path, new Path(outputfile));
        fs.close();
    }

    // 删除文件
    public static void deleteOutput(String deleteFile) throws IOException, InterruptedException {
        String hdfsOutput = hdfsUri + deleteFile;
        Path path = new Path(hdfsOutput);
        FileSystem fs = FileSystem.get(URI.create(hdfsUri), conf, "root");
        fs.deleteOnExit(path);
        fs.close();
    }

    // 重命名
    public static void moveFileName(String oldName, String newName) throws IOException, InterruptedException {
        String hdfsOldFile = hdfsUri + oldName;
        String hdfsNewFile = hdfsUri + newName;
        Path oldPath = new Path(hdfsOldFile);
        Path newPath = new Path(hdfsNewFile);
        FileSystem fs = FileSystem.get(URI.create(hdfsUri), conf, "root");
        fs.rename(oldPath, newPath);
        fs.close();
    }

    // 查看文件
    public static void listFile(String filePath) throws IOException, InterruptedException {
        String hdfsOldFile = hdfsUri + filePath;
        Path path = new Path(hdfsOldFile);
        FileSystem fs = FileSystem.get(URI.create(hdfsUri), conf, "root");
        RemoteIterator<LocatedFileStatus> locatedFileStatusRemoteIterator = fs.listFiles(path, true);
        while (locatedFileStatusRemoteIterator.hasNext()) {
            System.out.println("locatedFileStatusRemoteIterator = " + locatedFileStatusRemoteIterator);
        }
    }

    // 创建目录
    public static void mkdir(String dirPath) throws IOException, InterruptedException {
        String hdfsDirPath = hdfsUri + dirPath;
        Path path = new Path(hdfsDirPath);
        FileSystem fs = FileSystem.get(URI.create(hdfsUri), conf, "root");
        boolean result = fs.mkdirs(path);
        if (result) {
            System.out.println("目录" + dirPath + "已成功创建");
        }
    }
}
